/* The main component which represents DKS_COMM sub-system.
 * The users create an object of this class and use it.
 * The interface of the class is as follow:
 *
 * Create an object:
 * Users must pass THEIR OWN DKSRef object as a parameter.
 * ConnectionManager CM = new ConnectionManager( DKSRef );
 *
 * Sending a message:
 * First parameter: The DKSRef of the OTHER node (destination)
 * Second parameter: The Message object to be send
 * If the outputbuffer was full, it returns false. Otherwise it returns true.
 * NOTE: returning true does NOT mean the message has been sent successfully!!!!!
 * boolean Send( DKSRef, Message );
 *
 * Receiving a message:
 * It returns a BufferElement object.NOTE: it's a BLOCKING method.
 * It will be BLOCKED until some messages arrive.
 * BufferElement Receive();
 *
 */

package org.kth.dks.dks_comm;

import java.io.IOException;
import java.net.BindException;
import java.net.InetAddress;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;

import org.apache.log4j.Logger;
import org.kth.dks.dks_exceptions.DKSNodeAlreadyRegistered;
import org.kth.dks.dks_marshal.DKSMarshal;
import org.kth.dks.dks_marshal.MsgSrcDestWrapper;
import org.kth.dks.dks_node.DKSNode;
import org.kth.dks.util.CommunicationInfo;

public class ConnectionManager {
  private static ConnectionManager INSTANCE = null;
  private Logger log = Logger.getLogger(ConnectionManager.class);
  
  private boolean WEBSERVER = true;
  private Thread webServerThread = null;
  private WebServer webServer = null;
  
  Map nodes = Collections.synchronizedMap(new HashMap());
  
  private ConnectionHandler connHandler;
  private Listener listener = null; //This is a thread to listen to incomming connections
  
  private static class CommMode {
	  public static final CommMode BLOCKING = new CommMode("BLOCKING");
	  public static final CommMode NONBLOCKING = new CommMode("NONBLOCKING");
	  
	  private CommMode(String s) { name = s; } private final String name;
	  public String toString() { return name; }
  }
  
  private CommMode commMode = CommMode.BLOCKING;
  
  private DKSMarshal dksMarshal = null;
  private int port = 0;
  private long globalRTT = 1000 * 10;//10 sec

    /**
     * This static factory method is provided for backwards compatibility, and for testing purposes.
     * In practise, getInstance() should be used.
     * port 0 will bind to any port (use getLocalPort() to find out which)
     * @param port int port number to run on
     * @return ConnectionManager a new instance of a ConnectionManager
     */
    public synchronized static ConnectionManager newInstance(int port) throws IOException
    {
      return new ConnectionManager(port, null);
    }

    /**
     * Returns the sole connection handler registered with this instance of the ConnectionManager
     * @return ConnectionHandler
     */
    public ConnectionHandler getConnHandler() { return connHandler; }

  /**
   * Returns the sole instance of this class, port 0 will bind to any port (use getLocalPort() to find out which);
   * @return ConnectionManager
   */
  public synchronized static ConnectionManager getInstance(int port) throws IOException
  {
    if (INSTANCE==null) INSTANCE=new ConnectionManager(port, null); // lazy instantiation
    return INSTANCE;
  }

  /**
   * Returns the sole instance of this class, port 0 will bind to any port (use getLocalPort() to find out which);
   * @return ConnectionManager
   */
  public synchronized static ConnectionManager getInstanceMultiHome(int port, InetAddress bindAddr) throws IOException
  {
    if (INSTANCE==null) INSTANCE=new ConnectionManager(port, bindAddr); // lazy instantiation
    return INSTANCE;
  }

  /**
   * Returns the local port that this communication layer is actually listening to!
   * @return int port number
   */
  public synchronized int getLocalPort() {
    return listener.getLocalPort();
  }

  /**
   * Returns the hostname of the current ConnectionManager
   * @return String
   */
  public String getHostAddress() {
    return listener.getHostAddress();
  }

  /**
   * Creates a connection manager binded to a specific port and specific Bind address (0 if default)
   * @param _port int
   * @param bindAddr InetAddress
   * @throws IOException
   */
  private ConnectionManager( int _port, InetAddress bindAddr ) throws IOException
  {
    port = _port;
    dksMarshal  = new DKSMarshal(this);
    connHandler = new ConnectionHandler(dksMarshal, this);

    try {
    	if (commMode == CommMode.BLOCKING) {
    		if (bindAddr==null)
    			listener = new ListenerBlocking(port, this, dksMarshal, connHandler);
    		else
    			listener = new ListenerBlocking(port, this, dksMarshal, connHandler, bindAddr);
    	} else if (commMode == CommMode.NONBLOCKING) {
    		if (bindAddr==null)
    			listener = new ListenerNB(port, this, dksMarshal, connHandler);
    		else
    			listener = new ListenerNB(port, this, dksMarshal, connHandler, bindAddr);
    	}
       
       if (WEBSERVER) {
    	   int webServerPort = listener.getLocalPort()+1;
    	   webServer = new WebServer(this, webServerPort );
    	   webServerThread = new Thread(webServer, WebServer.class.getName());
    	   webServerThread.start();
       }

       //connGC = new ConnTableGC(this);
       //connGC.start();
    } catch (BindException be) {
      log.error( "DKS port ("+port+") already in use\n");
      throw be;
    } catch (IOException ioex) {
      throw ioex;
    }
  }

  public void registerDKSNode( DKSRef _dksRef, DKSNode node )  throws DKSNodeAlreadyRegistered
  {
	  nodes.put(_dksRef.getOverlayAddress(), node);
	  dksMarshal.registerDKSNode( _dksRef.getOverlayAddress() );
  }

  public void unregisterDKSNode( DKSRef _dksRef )
  {
    dksMarshal.unregisterDKSNode( _dksRef.getOverlayAddress() );
    nodes.remove(_dksRef.getDKSNetAddress());
  }
  
  public Collection getAllNodes() {
	  return nodes.values();
  }

  long getGlobalRTT() { return globalRTT; }

  void setGlobalRTT( long newRTT ) {
    globalRTT = newRTT;
    log.debug("New global RTT=" + globalRTT);
    //rttOutputFile.println(globalRTT);
  }

  public boolean send(MsgSrcDestWrapper triple)
  {
    if (triple.getDest() == null || triple.getSrc() == null) {
      return false;
    }

    if (commMode == CommMode.BLOCKING) {
    	return connHandler.send(triple);
    } else {
    	return ((ListenerNB)listener).send(triple);
    }
  }

  //Test!!!!!!!!!!!!!!!!
  public void register(Object obj) {
    //dispatcher.setObj( obj );
  }

  public DKSMarshal getDKSMarshal() {
    return dksMarshal;
  }

  public void end() {
    connHandler.end();
    listener.end();
    listener.interrupt();
    dksMarshal.end();
    //rttOutputFile.close();
  }

    public CommunicationInfo getComInfo(){
	int bRec        = connHandler.getBytesRec();
	int mRec        = connHandler.getMsgsRec();
	int bSent       = connHandler.getBytesSent();
	int mSend       = connHandler.getMsgsDelivered();
	int mUnacked    = (commMode == CommMode.BLOCKING ? 
                        connHandler.getMsgsUnacked() : 
                        connHandler.getMsgsUnackedNB());
	int mFailed     = connHandler.getMsgsFailed();
    int cons        = (commMode == CommMode.BLOCKING ? 
            connHandler.getNumConnections() :
            connHandler.getOpenConnections());

    int totalCons        = (commMode == CommMode.BLOCKING ? 
            connHandler.getNumConnections() :
            connHandler.getTotalConnections());
	return new CommunicationInfo(bRec,
				     mRec,
				     bSent,
				     mSend,
				     mUnacked,
				     mFailed,
				     cons,
                     totalCons);
    }

    public double getNodeRTT(DKSRef node) {
      return connHandler.getNodeRTT(node);
    }

	public WebServer getWebServer() {
		return webServer;
	}

} //ConnectionManager class
